package com.ruyuan.event.streaming.sink;
import com.google.protobuf.Message;
import com.googlecode.protobuf.format.JsonFormat;
import org.apache.flink.streaming.connectors.fs.StreamWriterBase;
import org.apache.flink.streaming.connectors.fs.Writer;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
import java.nio.charset.Charset;
import java.nio.charset.IllegalCharsetNameException;
import java.nio.charset.UnsupportedCharsetException;

public class ProtoBufToStringWriter<T extends Message> extends StreamWriterBase<T> {
    private static final long serialVersionUID = 1L;

    private String charsetName;

    private transient Charset charset;

    public ProtoBufToStringWriter() {
        this("UTF-8");
    }

    public ProtoBufToStringWriter(String charsetName) {
        this.charsetName = charsetName;
    }

    protected ProtoBufToStringWriter(ProtoBufToStringWriter<T> other) {
        super(other);
        this.charsetName = other.charsetName;
    }


    @Override
    public void open(FileSystem fs, Path path) throws IOException {
        super.open(fs, path);

        try {
            this.charset = Charset.forName(charsetName);
        }
        catch (IllegalCharsetNameException e) {
            throw new IOException("The charset " + charsetName + " is not valid.", e);
        }
        catch (UnsupportedCharsetException e) {
            throw new IOException("The charset " + charsetName + " is not supported.", e);
        }
    }

    //日志类型转换为Json格式的String
    @Override
    public void write(T element) throws IOException {
        FSDataOutputStream outputStream = getStream();
        outputStream.write(JsonFormat.printToString(element).getBytes(charset));
        outputStream.write('\n');
    }

    @Override
    public Writer<T> duplicate() {
        return new ProtoBufToStringWriter<>(this);
    }
}

